package com.zanglikun.rocketmq.transaction;

import org.apache.commons.lang3.StringUtils;
import org.apache.rocketmq.client.producer.*;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageExt;

/**
 * @Author: zanglikun
 * @Date: 2022/4/1 14:17
 * @Description:
 */
public class TransProducer {
    public static void main(String[] args) throws Exception {
        // todo 1 创建消息生产者对象,同时指定生产者组名
        TransactionMQProducer producer = new TransactionMQProducer("group5");
        // todo 2 指定NameServer地址
        producer.setNamesrvAddr("192.168.253.128:9876;192.168.253.129:9876");
        // todo 3 设定消息监听器
        producer.setTransactionListener(new TransactionListener() {
            // 本地事务
            @Override
            public LocalTransactionState executeLocalTransaction(Message message, Object o) {
                if (StringUtils.endsWith("TAGA",message.getTags())){
                    return LocalTransactionState.COMMIT_MESSAGE;    // 提交事务
                }else if (StringUtils.endsWith("TAGB",message.getTags())){
                    return LocalTransactionState.ROLLBACK_MESSAGE;  // 执行回滚
                }else if (StringUtils.endsWith("TAGC",message.getTags())){
                    return LocalTransactionState.UNKNOW;            // 使得被事务回查发现
                }
                return LocalTransactionState.UNKNOW;
            }


            // MQ进行消息事务的回查，当状态为UNKNOW的时候才会触发执行。注意，当状态是ROLLBACK_MESSAGE是不会触发执行的
            @Override
            public LocalTransactionState checkLocalTransaction(MessageExt messageExt) {
                System.out.println("消息回查："+messageExt.getTags());
                return LocalTransactionState.COMMIT_MESSAGE;
            }
        });
        // todo 4 启动生产者
        producer.start();
        // todo 定义tags
        String[] tags = new String[]{"TAGA","TAGB","TAGC"};
        for (int i = 0; i < 3; i++) {
            // todo 5 创建消息对象,同时指定topic、tags、body
            Message message = new Message("TransTopic", tags[i], ("你好"+(i+1)).getBytes());
            // todo 6 发送消息带有事务
            SendResult sendResult = producer.sendMessageInTransaction(message,null);
            // todo 6 处理发送结果
            SendStatus sendStatus = sendResult.getSendStatus();
            String msgId = sendResult.getMsgId();
            int queueId = sendResult.getMessageQueue().getQueueId();
            System.out.println("消息发送状态" + sendStatus + ",消息Id:" + msgId + ",消息队列Id" + queueId);
            Thread.sleep(100);
        }
        // todo 7 不得关闭生产者
        // producer.shutdown();
    }
}
